use crate::utils::access_guard::AccessGuard;
use crate::utils::single_linked_list::{SingleLinkedList, SingleLinkedNode};
use std::cell::Cell;
use std::ptr::{null_mut, NonNull};
use std::sync::atomic::{AtomicPtr, Ordering};

#[derive(Debug)]
pub(crate) struct AtomicSingleLinkedList<T>
where
    T: SingleLinkedNode<T>,
{
    head: AtomicPtr<T>,
}

impl<T> Default for AtomicSingleLinkedList<T>
where
    T: SingleLinkedNode<T>,
{
    fn default() -> Self {
        Self {
            head: AtomicPtr::new(null_mut()),
        }
    }
}

impl<T> AtomicSingleLinkedList<T>
where
    T: SingleLinkedNode<T>,
{
    #[cfg(test)]
    pub(crate) fn first(&self) -> Option<NonNull<T>> {
        NonNull::new(self.head.load(Ordering::Relaxed))
    }
    pub(crate) fn take(&self) -> SingleLinkedList<T> {
        SingleLinkedList {
            head: Cell::new(NonNull::new(self.head.swap(null_mut(), Ordering::SeqCst))),
        }
    }

    pub(crate) fn push<G, N>(&self, access_guard: &G, node: N)
    where
        N: Into<NonNull<T>>,
        G: AccessGuard<T>,
    {
        let node_ptr = node.into();
        let node = access_guard.access(node_ptr);
        debug_assert!(node.next().is_none());
        let mut old_head = null_mut();
        self.head
            .fetch_update(Ordering::Relaxed, Ordering::SeqCst, |head_node| {
                old_head = head_node;
                Some(node_ptr.as_ptr())
            })
            .expect("should never fail");
        *node.next() = NonNull::new(old_head);
    }
}
#[cfg(test)]
mod tests {
    use crate::utils::access_guard::AccessGuard;
    use crate::utils::atomic_single_linked_list::AtomicSingleLinkedList;
    use crate::utils::single_linked_list::{SingleLinkedList, SingleLinkedNode};
    use std::ptr::NonNull;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::Arc;
    use std::time::Duration;
    use std::{iter, thread};

    #[derive(Debug)]
    struct TestNode {
        val: usize,
        next: Option<NonNull<TestNode>>,
    }

    impl SingleLinkedNode<TestNode> for TestNode {
        fn next(&mut self) -> &mut Option<NonNull<TestNode>> {
            &mut self.next
        }
    }

    struct TestAccessGuard;

    impl AccessGuard<TestNode> for TestAccessGuard {
        fn before_access(&self, _ptr: NonNull<TestNode>) {}
    }

    fn assert_list<T, G>(
        guard: &G,
        list: &AtomicSingleLinkedList<T>,
        nodes: &Vec<T>,
        expected: &[usize],
    ) where
        G: AccessGuard<T>,
        T: SingleLinkedNode<T>,
    {
        let mut node_idx = 0;
        let mut current = NonNull::new(list.head.load(Ordering::Relaxed));
        while let Some(node_ptr) = current {
            let node = guard.access(node_ptr);
            assert_eq!(node_ptr, NonNull::from(&nodes[expected[node_idx]]));
            node_idx += 1;
            current = *node.next();
        }
        assert_eq!(node_idx, expected.len());
        assert_eq!(
            expected.is_empty(),
            list.head.load(Ordering::Relaxed).is_null()
        );
    }
    #[test]
    fn test_atomic_single_linked_list() {
        let guard = TestAccessGuard;
        let nodes = iter::repeat(0)
            .take(5)
            .map(|_| TestNode { val: 0, next: None })
            .collect::<Vec<_>>();

        let list = AtomicSingleLinkedList::default();
        assert_list(&guard, &list, &nodes, &[]);

        // Push many nodes
        nodes.iter().for_each(|node| {
            let old_head = list.head.load(Ordering::Relaxed);
            list.push(&guard, node);
            assert_eq!(
                list.head.load(Ordering::Relaxed),
                NonNull::from(node).as_ptr()
            );
            assert_eq!(NonNull::new(old_head), node.next);
        });
        assert_list(&guard, &list, &nodes, &[4, 3, 2, 1, 0]);
        assert_eq!(list.take().iter(&guard).count(), 5);
        assert_eq!(list.take().iter(&guard).count(), 0);
    }

    unsafe impl<T> Send for SingleLinkedList<T> where T: SingleLinkedNode<T> {}
    /// 对 AtomicSingleLinkedList 进行并发操作测试
    /// 起一定数量的线程，在线程中构造带有序号的节点，并发 push，和 take，
    /// 所有线程结束后，收集 take 的结果，汇总并排序，检查是否丢节点。
    #[test]
    fn test_atomic_single_linked_list_multi_thread() {
        const THREAD_NUMBER: usize = 128;
        const NODE_PRE_THREAD: usize = 2000;
        let list = Arc::new(AtomicSingleLinkedList::default());
        let thread_started = Arc::new(AtomicUsize::new(0));
        let handles = (0..THREAD_NUMBER)
            .map(|id| {
                let list = list.clone();
                let thread_started = thread_started.clone();
                thread::spawn(move || {
                    thread_started.fetch_add(1, Ordering::Relaxed);
                    let take_step = id * 3 + 10;
                    let val_start = id * NODE_PRE_THREAD;
                    let guard = TestAccessGuard;
                    let mut taken_from_list = vec![];
                    for i in 0..NODE_PRE_THREAD {
                        let i = i + val_start;
                        let node = Box::new(TestNode { val: i, next: None });
                        list.push(&guard, &*node);
                        Box::leak(node);
                        if i % take_step == 0 {
                            taken_from_list.push(list.take());
                        }
                    }
                    taken_from_list.push(list.take());
                    taken_from_list
                })
            })
            .collect::<Vec<_>>();
        // 等待所有线程启动
        while thread_started.load(Ordering::Relaxed) < THREAD_NUMBER {
            println!("Thread started: {}", thread_started.load(Ordering::Relaxed));
            thread::sleep(Duration::from_millis(100));
        }
        println!("Thread started: {}", thread_started.load(Ordering::Relaxed));
        let guard = TestAccessGuard;
        let mut nodes = handles
            .into_iter()
            .map(|h| h.join().unwrap())
            .flatten()
            .map(|x| x.iter(&guard))
            .flatten()
            .map(|x| unsafe { Box::from_raw(x.as_ptr()) })
            .collect::<Vec<_>>();
        assert_eq!(nodes.len(), THREAD_NUMBER * NODE_PRE_THREAD);
        nodes.sort_by(|a, b| a.val.cmp(&b.val));
        for (i, node) in nodes.into_iter().enumerate() {
            assert_eq!(i, node.val);
        }
    }
}
